﻿using EasyNetQ;
using EasyNetQ.Topology;
using Newtonsoft.Json;
using System.Text;
namespace Magic.Rmq;

public class RmqService
{
    private static readonly Lazy<IRmqFactory> Factory = new(() => new RmqFactory(), LazyThreadSafetyMode.ExecutionAndPublication);
    public static IBus GetClient(string connectionStringOrName)
    {

        return Factory.Value.Create(connectionStringOrName);
    }

    public static IAdvancedBus GetAdvancedClient(string connectionStringOrName)
    {
        return Factory.Value.Create(connectionStringOrName).Advanced;
    }

    public static IExchange GetExchange(string connectionStringOrName)
    {
        return Factory.Value.GetExchange(connectionStringOrName);
    }

    public static IPullingConsumer<PullResult> GetPullingConsumer(string connectionStringOrName, string queueName)
    {
        return Factory.Value.GetPullingConsumer(connectionStringOrName, queueName);
    }

    public static async Task EnqueueAsync<T>(T entity, string routingKey, PriorityLevel level = PriorityLevel.Normal, string connectionStringOrName = "")
    {
        if (entity == null)
            return;
        var bus = GetAdvancedClient(connectionStringOrName);

        var exchange = GetExchange(connectionStringOrName);
        await bus.PublishAsync(exchange, routingKey, false, new Message<T>(entity)
        {
            Properties =
            {
                DeliveryMode=(byte)2,
                Priority = (byte)Convert.ToInt32(level),
            }
        });
    }

    public static void Enqueue<T>(T entity, string routingKey, PriorityLevel level = PriorityLevel.Normal, string connectionStringOrName = "")
    {
        if (entity == null)
            return;
        var bus = GetAdvancedClient(connectionStringOrName);

        var exchange = GetExchange(connectionStringOrName);
        bus.Publish(exchange, routingKey, false, new Message<T>(entity)
        {
            Properties =
            {
                Priority = (byte)Convert.ToInt32(level),
            }
        });
    }

    /// <summary>
    /// 获取一条消息，没有数据时，返回默认值
    /// </summary>
    /// <typeparam name="T"></typeparam>
    /// <param name="queueName"></param>
    /// <param name="connectionStringOrName"></param>
    /// <returns></returns>
    public static async Task<T> GetOneMessageAsync<T>(string queueName, string connectionStringOrName = "")
    {
        var consumer = GetPullingConsumer(connectionStringOrName, queueName);
        var result = await consumer.PullAsync();
        try
        {
            var message = Encoding.UTF8.GetString(result.Body);
            return JsonConvert.DeserializeObject<T>(message);
        }
        catch (Exception)
        {
            return default;
        }
    }

    /// <summary>
    /// 批量获取消息，没有数据时，返回默认值
    /// </summary>
    /// <typeparam name="T"></typeparam>
    /// <param name="queueName"></param>
    /// <param name="batchSize"></param>
    /// <param name="connectionStringOrName"></param>
    /// <returns></returns>
    public static async Task<List<T>> GetBatchMessageAsync<T>(string queueName, int batchSize,
        string connectionStringOrName = "")
    {
        var consumer = GetPullingConsumer(connectionStringOrName, queueName);
        var result = await consumer.PullBatchAsync(batchSize);
        var lstData = new List<T>();
        foreach (var item in result.Messages)
        {
            try
            {
                var message = Encoding.UTF8.GetString(item.Body);
                lstData.Add(JsonConvert.DeserializeObject<T>(message));
            }
            catch (Exception ex)
            {
                continue;
            }
        }
        return lstData;
    }

    /// <summary>
    /// 循环阻塞获取消息，直到取到数据为止
    /// </summary>
    /// <typeparam name="T"></typeparam>
    /// <param name="queueName"></param>
    /// <param name="connectionStringOrName"></param>
    /// <returns></returns>
    public static async Task<T> GetOneMessageLoopAsync<T>(string queueName, string connectionStringOrName = "")
    {
        while (true)
        {
            var consumer = GetPullingConsumer(connectionStringOrName, queueName);
            var result = await consumer.PullAsync();
            if (result.IsAvailable)
            {
                try
                {
                    var message = Encoding.UTF8.GetString(result.Body);
                    return JsonConvert.DeserializeObject<T>(message);
                }
                catch (Exception ex)
                {
                    continue;
                }
            }
            await Task.Delay(TimeSpan.FromSeconds(2));
        }
    }

    /// <summary>
    /// 批量获取消息，满足batchSize或者最大等待时长waitTime返回数据
    /// </summary>
    /// <typeparam name="T"></typeparam>
    /// <param name="queueName"></param>
    /// <param name="batchSize"></param>
    /// <param name="waitTime"></param>
    /// <param name="connectionStringOrName"></param>
    /// <returns></returns>
    public static async Task<List<T>> GetBatchMessageLoopAsync<T>(string queueName, int batchSize, TimeSpan? waitTime,
        string connectionStringOrName = "")
    {
        var lstData = new List<T>();
        var begin = DateTime.Now;
        while (lstData.Count < batchSize && DateTime.Now - begin < waitTime || waitTime == null)
        {
            var consumer = GetPullingConsumer(connectionStringOrName, queueName);
            var result = await consumer.PullBatchAsync(batchSize);
            foreach (var item in result.Messages)
            {
                try
                {
                    var message = Encoding.UTF8.GetString(item.Body);
                    lstData.Add(JsonConvert.DeserializeObject<T>(message));
                }
                catch
                {
                }
            }
        }
        return lstData;
    }



    public static async Task<int> GetQueueCountAsync(string queueName, string connectionStringOrName = "")
    {
        var stats = await GetAdvancedClient(connectionStringOrName).GetQueueStatsAsync(new Queue(queueName));
        return (int)stats.MessagesCount;
    }

    public static int GetQueueCount(string queueName, string connectionStringOrName = "")
    {
        var stats = GetAdvancedClient(connectionStringOrName).GetQueueStats(new Queue(queueName));
        return (int)stats.MessagesCount;
    }
}

public abstract class RmqConsumerWorkerBase<T>
{
    protected int ConsumerCount { get; set; } = 50;
    protected string QueueName { get; set; }
    protected string ConnectionStringOrName { get; set; } = RmqOptions.ConnectionString;
    private IAdvancedBus _bus;
    public Task Start()
    {
        _bus = RmqService.GetAdvancedClient(ConnectionStringOrName);
        _bus.Disconnected += this._bus_Disconnected;
        _bus.Consume(new Queue(QueueName), async (body, properties, info) =>
          {
              try
              {

                  var message = Encoding.UTF8.GetString(body);
                  T dto = JsonConvert.DeserializeObject<T>(message);
                  await HandleComingMessage(dto);

              }
              catch (Exception ex)
              {
                  Console.WriteLine($"Consume异常：{ex}");
              }
          }, configure =>
          {
              configure.WithPrefetchCount((ushort)ConsumerCount);
          });
        return Task.CompletedTask;

    }

    private void _bus_Disconnected(object sender, DisconnectedEventArgs e)
    {
        Console.WriteLine($"{QueueName}断开连接：{e.Reason}");
        // this.Start();
    }

    public Task StartWithAck(bool requeue = false, Func<ulong, Task<bool>> retryLimitFun = null)
    {
        var _bus = RmqService.GetAdvancedClient(ConnectionStringOrName);

        _bus.Consume(new Queue(QueueName), async (body, properties, info) =>
        {
            try
            {
                var message = Encoding.UTF8.GetString(body);
                T dto = JsonConvert.DeserializeObject<T>(message);
                await HandleComingMessage(dto);
                return EasyNetQ.Consumer.AckStrategies.Ack;
            }
            catch (RmqQueueNoException ex)
            {
                return EasyNetQ.Consumer.AckStrategies.Ack;
            }
            catch (RmqQueueException ex)
            {
                if (retryLimitFun != null)
                {
                    var result = await retryLimitFun.Invoke(info.DeliveryTag);
                    if (!result)
                    {
                        // 直接消费，不在重新入队；
                        return EasyNetQ.Consumer.AckStrategies.Ack;
                    }
                }
                return EasyNetQ.Consumer.AckStrategies.NackWithoutRequeue;
            }
            catch (Exception ex)
            {
                if (retryLimitFun != null)
                {
                    var result = await retryLimitFun.Invoke(info.DeliveryTag);
                    if (!result)
                    {
                        // 直接消费，不在重新入队；
                        return EasyNetQ.Consumer.AckStrategies.Ack;
                    }
                }
                Console.WriteLine($"Consume异常：{ex}");
                return requeue ? EasyNetQ.Consumer.AckStrategies.NackWithRequeue : EasyNetQ.Consumer.AckStrategies.NackWithoutRequeue;
            }
        }, configure =>
        {
            configure.WithPrefetchCount((ushort)ConsumerCount);
        });
        return Task.CompletedTask;
    }


    public Task StartWithAck(Func<T, Task<bool>> retryLimitFun, bool requeue = false)
    {
        var _bus = RmqService.GetAdvancedClient(ConnectionStringOrName);

        _bus.Consume(new Queue(QueueName), async (body, properties, info) =>
        {
            T dto = default;
            try
            {
                var message = Encoding.UTF8.GetString(body);
                dto = JsonConvert.DeserializeObject<T>(message);
                await HandleComingMessage(dto);
                return EasyNetQ.Consumer.AckStrategies.Ack;
            }
            catch (RmqQueueNoException ex)
            {
                return EasyNetQ.Consumer.AckStrategies.Ack;
            }
            catch (RmqQueueException ex)
            {
                if (retryLimitFun != null)
                {
                    var result = await retryLimitFun.Invoke(dto);
                    if (!result)
                    {
                        // 直接消费，不在重新入队；
                        return EasyNetQ.Consumer.AckStrategies.Ack;
                    }
                }
                return EasyNetQ.Consumer.AckStrategies.NackWithoutRequeue;
            }
            catch (Exception ex)
            {
                if (retryLimitFun != null)
                {
                    var result = await retryLimitFun.Invoke(dto);
                    if (!result)
                    {
                        // 直接消费，不在重新入队；
                        return EasyNetQ.Consumer.AckStrategies.Ack;
                    }
                }
                Console.WriteLine($"Consume异常：{ex}");
                return requeue ? EasyNetQ.Consumer.AckStrategies.NackWithRequeue : EasyNetQ.Consumer.AckStrategies.NackWithoutRequeue;
            }
        }, configure =>
        {
            configure.WithPrefetchCount((ushort)ConsumerCount);
        });
        return Task.CompletedTask;
    }


    protected abstract Task HandleComingMessage(T dto);
}

/// <summary>
/// 拒绝消息处理
/// </summary>
public class RmqQueueException : Exception
{
    public RmqQueueException(string msg) : base(msg)
    {

    }
}

/// <summary>
/// 同意消息已处理
/// </summary>
public class RmqQueueNoException : Exception
{
    public RmqQueueNoException(string msg) : base(msg)
    {

    }
}
